package cn.com.rocketmq.async;

import org.apache.rocketmq.client.consumer.DefaultLitePullConsumer;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import java.util.List;

/**
 * 随机获取消息队列
 * @author xuhaixiang
 * @date 2023-09-10
 */
public class LitePullConsumer {

    /**
     * 日志对象
     */
    private static  final Logger logger = LoggerFactory.getLogger(LitePullConsumer.class);

    public static void main(String[] args) {
        DefaultLitePullConsumer consumer = null;
        try {
            consumer = new DefaultLitePullConsumer("group_rocketmq_asyncDemo");
            consumer.setNamesrvAddr("192.168.113.101:9876");
            consumer.subscribe("rocketmq_asyncDemo", "*");
            consumer.start();
            while (true) {
                List<MessageExt> messageExtList = consumer.poll();
                messageExtList.forEach(item -> {
                    try {
                        logger.info("获取消息成功！消息队列ID={}，消息ID={}，消息内容{}", item.getQueueId(),item.getMsgId(), new String(item.getBody(), "UTF-8"));
                    } catch (Exception e) {
                        logger.error("获取消息异常！",e);
                    }
                });
            }
        } catch (MQClientException e) {
            logger.error("获取消息异常！",e);
        } finally {
            if (consumer != null) {
                consumer.shutdown();
            }
        }
    }
}
